package peer;

import java.rmi.RemoteException;
import java.util.concurrent.LinkedBlockingQueue;

import protocols.Protocol;
import protocols.joinProtocol.JoinProtocol;

import javaspaces.Locator;
import net.jini.core.entry.UnusableEntryException;
import net.jini.core.lease.Lease;
import net.jini.core.transaction.TransactionException;
import net.jini.space.JavaSpace;

public class JavaspaceConnection extends Connection{

	public static final long TIMEOUT = 150000; // 15 seconds
	
	JavaSpace space;
	Message template;
	private String rootId = null;
	
	public JavaspaceConnection(String ip, Peer _callback) {
		super(_callback);
		System.out.println("Connecting to " + ip);
		space = Locator.getSpace(ip);
	}
	
	public void init() {
		Message<String> leaderTemplate = new Message<String>();
		Message<String> entry;
		leaderTemplate.protocol = "Leader";
		leaderTemplate.type = JoinProtocol.MessageTypes.leader.toString();
			

		try {
			System.out.println("Checking for existing leaders"); System.out.flush();
			System.out.println("why is the space null: " + (space == null));
			entry = (Message) space.read(leaderTemplate, null, Lease.DURATION);
			if (entry == null) {
				System.out.println("alas...");
				leaderTemplate.id = Peer.id;
				space.write(leaderTemplate, null, TIMEOUT);
				this.becomeLeader();

							
			} else {
				System.out.println("Leader found - " + entry);	
				entry.type = JoinProtocol.MessageTypes.join.toString();
				entry.protocol = JoinProtocol.NAME;
				entry.data = Peer.id;
				System.out.println("Posting - " + entry);
				space.write(entry, null, Lease.FOREVER);

				if (Peer.id.equals("2")) {
					callback.test();
				}
			}
		} catch (RemoteException e) {
			e.printStackTrace();
		} catch (UnusableEntryException e) {
			e.printStackTrace();
		} catch (TransactionException e) {
			e.printStackTrace();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
	
	@Override
	public Object read(String protocol) {
		try {
			System.out.println("Checking for messages under " + protocol);
			LinkedBlockingQueue q =  msgQueue.get(protocol);
			System.out.println(q == null);
			return q.take();
		} catch (InterruptedException e) {
			return read(protocol);
		} catch (NullPointerException e) {
			System.err.println("Unsupported protocol. Can't return message.");
			return null;
		}
		
	}

	@Override
	public void write(Object o) {
	
		Message<String> m = (Message<String>) o;
		System.out.println("Sending... " + m);
		try {
			space.write(m, null, Lease.FOREVER);
		} catch (RemoteException e) {
			e.printStackTrace();
		} catch (TransactionException e) {
			e.printStackTrace();
		}
		
	}

	@Override
	public void run() {
		init();
		System.out.println(protocols.keySet());
		for (String protocol : protocols.keySet()) {
				
			
			template = new Message();			
			template.id = Peer.id;
			template.protocol = protocol;
			
			new Thread(new Runnable() {
				public void run() {
					while(true) {
						Message m = null;
						try {
							System.out.println("Looking for " + template);
							m = (Message) space.take(template, null, Lease.FOREVER);
							if (protocols.keySet().contains(m.protocol)) {
								msgQueue.get(m.protocol).put(m);
								System.out.println("message added to queue...\n\t"+m);
							} else {
								System.out.println("Unsupported protocol.  message discard...");
							}														
						} catch (RemoteException e) {
							e.printStackTrace();
						} catch (UnusableEntryException e) {
							e.printStackTrace();
						} catch (TransactionException e) {
							e.printStackTrace();
						} catch (InterruptedException e) {
							e.printStackTrace();
						}
						System.out.println(m);
					}
				}
			}).start();
		}
	}
	
	public void close() {
		for (Protocol protocol : protocols.values()) {
			protocol.stop();
		}		
		Message<String> m = new Message<String>();
		m.id = Peer.id;
		m.protocol = "Leader";
		m.type = JoinProtocol.MessageTypes.leader.toString();
		try {
			space.take(m, null, Lease.FOREVER);
		} catch (RemoteException e) {
			e.printStackTrace();
		} catch (UnusableEntryException e) {
			e.printStackTrace();
		} catch (TransactionException e) {
			e.printStackTrace();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		space = null;
	}
	
	private void becomeLeader() {
		
		protocols.get(JoinProtocol.NAME).setState(JoinProtocol.States.leader);
		callback.becomeLeader();			
	}

}
